package com.spider.dao.hbasefilter;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class HBaseFilterImp implements HBaseFilter {
	Connection connection = null;
	Table table = null;
	Configuration conf = null;

	public HBaseFilterImp() {
		conf = new Configuration();
		conf = HBaseConfiguration.create();
		conf.set("hbase.rootdir", "hdfs://master1:9000/hbase");
		conf.set("hbase.zookeeper.quorum", "slave1,slave2");
		try {
			connection = ConnectionFactory.createConnection(conf);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	@Override
	public List<Result> ColumnPaginationFilter(String tableName, int limit, int offset) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new ColumnPaginationFilter(limit, offset);
			Scan scan = new Scan();
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	@Override
	public List<Result> ColumnPrefixFilter(String tableName, String prefix) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new ColumnPrefixFilter(Bytes.toBytes(prefix));
			Scan scan = new Scan();
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	@Override
	public List<Result> ColumnRangeFilter(String tableName, String minColumn, boolean minColumnInclusive,
										  String maxColumn, boolean maxColumnInclusive) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new ColumnRangeFilter(Bytes.toBytes(minColumn), minColumnInclusive,
					Bytes.toBytes(maxColumn), maxColumnInclusive);

			Scan scan = new Scan();
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}
	@Override
	public List<Result> ColumnRangeFilter(String tableName, String minColumn, boolean minColumnInclusive,
										  String maxColumn, boolean maxColumnInclusive, String startRow, String stopRow) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new ColumnRangeFilter(Bytes.toBytes(minColumn), minColumnInclusive,
					Bytes.toBytes(maxColumn), maxColumnInclusive);

			Scan scan = new Scan().setStartRow(Bytes.toBytes(startRow)).setStopRow(Bytes.toBytes(stopRow))
					.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	@Override
	public List<Result> FirstKeyOnlyFilter(String tableName) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new FirstKeyOnlyFilter();

			Scan scan = new Scan();
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	@Override
	public List<Result> RowFilter(String tableName, String rowkey) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new RowFilter(CompareFilter.CompareOp.LESS_OR_EQUAL, new BinaryComparator(Bytes.toBytes(rowkey)));

			Scan scan = new Scan();
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	@Override
	public List<Result> PrefixFilter(String tableName, String rowLike) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new PrefixFilter(Bytes.toBytes(rowLike));

			Scan scan = new Scan();
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	@Override
	public List<Result> KeyOnlyFilter(String tableName) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new KeyOnlyFilter();

			Scan scan = new Scan();
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	@Override
	public List<Result> RandomRowFilter(String tableName, float chance) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new RandomRowFilter(chance);

			Scan scan = new Scan();
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	@Override
	public List<Result> ValueFilter(String tableName, String value) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new ValueFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(value));

			Scan scan = new Scan();
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	@Override
	public List<Result> RowFilter(String tableName, String rowkey, String family, String qualifier) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes(rowkey)));

			Scan scan = new Scan();
			scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	@Override
	public List<Result> RowRegexFilter(String tableName, String rowkeyRegex) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(rowkeyRegex));

			Scan scan = new Scan();
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	@Override
	public List<Result> RowSubFilter(String tableName, String rowkeySubString) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new SubstringComparator(rowkeySubString));

			Scan scan = new Scan();
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	@Override
	public List<Result> FamilyFilter(String tableName, String family) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new FamilyFilter(CompareFilter.CompareOp.EQUAL,new BinaryComparator(Bytes.toBytes(family)));

			Scan scan = new Scan();
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	@Override
	public List<Result> QualifierFilter(String tableName, String qualifier) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new QualifierFilter(CompareFilter.CompareOp.EQUAL,new BinaryComparator(Bytes.toBytes(qualifier)));


			Scan scan = new Scan();
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	@Override
	public List<Result> KeyOnlyFilter(String tableName, String family, String qualifier) {
		List<Result> list = null;
		ResultScanner scanner = null;
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Filter filter = new KeyOnlyFilter();

			Scan scan = new Scan();
			scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
			scan.setFilter(filter);
			scanner = table.getScanner(scan);
			list = new ArrayList<Result>();
			for (Result rs : scanner) {
				list.add(rs);
			}
		} catch (Exception e) {
			scanner.close();
		}
		return list;
	}

	@Override
	public Result lastRowKeyFilter(String tableName){
		List<Result> list = null;
		ResultScanner scanner = null;
		Result lastResult = null;
		try{
			Table table = connection.getTable(TableName.valueOf(tableName));
			Scan scan = new Scan();
			scan.setReversed(true);
			scanner = table .getScanner(scan);
			for(Result result : scanner){
				lastResult = result;
				break;
			}
		}catch(Exception e){
			scanner.close();
		}
		return lastResult;
	}
}